/*
 * Copyright (c) 2017,安徽皖通科技股份有限公司 All rights reserved.
 */
package com.wttech.dzzf.process.service.upload;

import cn.com.taiji.sdk.comm.ETCCommHelper;
import cn.com.taiji.sdk.model.comm.protocol.AbstractSdtsRequest;
import cn.com.taiji.sdk.model.comm.protocol.AbstractSdtsResponse;
import cn.com.taiji.sdk.model.comm.protocol.transaction.RechargeUploadRequest;
import cn.com.taiji.sdk.model.comm.protocol.user.CardUploadRequest;
import cn.com.taiji.sdk.model.comm.protocol.user.OBUUploadRequest;
import cn.com.taiji.sdk.model.comm.protocol.user.UserUploadRequest;
import cn.com.taiji.sdk.model.comm.protocol.user.VehicleUploadRequest;
import com.wttech.dzzf.process.constant.ExecuteTable;
import com.wttech.dzzf.process.constant.UserConstants;
import com.wttech.dzzf.process.dao.TransferUserInfoDao;
import com.wttech.dzzf.process.service.user.UserDetailInfosCreatorService;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

import static com.wttech.dzzf.process.constant.UserConstants.*;

/**
 * 上传数据服务实现
 *
 * @author MaShibo
 */
@Service
public class UploadRequestInfos {


    private static final Logger LOGGER = LoggerFactory.getLogger(UploadRequestInfos.class);

    /**
     * 默认每次处理数量
     */
    private static final int DEFAULT_PROCESS_NUM = 10000;

    private static final String DEFAULT_SENDER = "61";

    /**
     * 待上传数据表
     */
    private List<ExecuteTable> executeTables;

    /**
     * 是否限制处理数量
     */
    private boolean isAssigned = false;

    /**
     * 是否后续处理
     */
    private boolean isAfterProcess = true;


    @Resource
    private ThreadPoolTaskExecutor executor;

    @Resource
    private UserDetailInfosCreatorService userDetailInfosCreatorService;

    @Resource
    private TransferUserInfoDao transferUserInfoDao;

    private void uploadReqInfos() {
        if (executeTables == null || executeTables.isEmpty()) {
            LOGGER.warn("未指定待上传数据对象");
            return;
        }
        int coreNum = executor.getCorePoolSize();
        LOGGER.info("线程池核心池数量{}", coreNum);
        List<String> tableNameList = executeTables.stream().map(ExecuteTable::getTableName).collect(Collectors.toList());
        String tableNames = StringUtils.join(tableNameList, ", ");
        LOGGER.info("待上传数据表：{}", tableNames);
        executeTables.forEach(table -> {
            int totalNum = transferUserInfoDao.queryUserInfoCount(table.getTableName());
            LOGGER.info("{}共{}条数据待处理...", table.getTableName(), totalNum);
            //被限制的上传数量
            int limitedNum = totalNum < DEFAULT_PROCESS_NUM * 50 ? totalNum : totalNum / 10;
            //实际处理数量
            int total = isAssigned ? limitedNum : totalNum;
            LOGGER.info("实际处理{}条数据", total);
            int processNum = total / DEFAULT_PROCESS_NUM > coreNum ? DEFAULT_PROCESS_NUM : (int) Math.ceil((double) total / coreNum);
            LOGGER.info("每次处理{}条数据", processNum);
            int count = (int) Math.ceil((double) total / processNum);
            LOGGER.info("共需处理{}次", count);
            List<Future<Void>> futures = new ArrayList<>();
            for (int i = 0; i < count; i++) {
                Future<Void> future = executor
                        .submit(new SubProcessThread(i + 1, count, processNum, table));
                futures.add(future);
            }
            userDetailInfosCreatorService.getFutureResult(futures);
        });
        executor.shutdown();
    }

    private void setExecuteTables(List<ExecuteTable> executeTables) {
        this.executeTables = executeTables;
    }

    private void setAssigned(boolean assigned) {
        isAssigned = assigned;
    }

    public class SubProcessThread implements Callable<Void> {

        private int count;

        private int total;

        private int processNum;

        private ExecuteTable table;

        public SubProcessThread(int count, int total, int processNum, ExecuteTable table) {
            this.count = count;
            this.total = total;
            this.processNum = processNum;
            this.table = table;
        }

        @Override
        public Void call() throws Exception {
            LOGGER.info("【{}】第{}-{}次，开始执行...", new Object[]{table.getStorePath(), total, count});
            int startRow = (count - 1) * processNum + 1;
            int endRow = count * processNum;
            List<? extends AbstractSdtsRequest> infos = transferUserInfoDao.queryUseriInfoRecord(table.getTableName(), table.getRequestType(),
                    startRow, endRow);
            infos.forEach(info -> {
                try {
                    uploadSdtsInfos(info);
                } catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                    if (isAfterProcess) {
                        uploadAgain(info);
                    }
                }
            });
            LOGGER.info("【{}】第{}-{}次，结束执行...", new Object[]{table.getStorePath(), total, count});
            return null;
        }

        /**
         * 上传数据
         *
         * @param info 请求信息
         * @throws IOException IO异常
         */
        private void uploadSdtsInfos(AbstractSdtsRequest info) throws IOException {
            AbstractSdtsResponse response = ETCCommHelper.upload(info.getFileName(DEFAULT_SENDER), info,
                    table.getResponseType());
            LOGGER.info("【{}】第{}-{}次正在执行，响应信息: {}，响应时间: {}",
                    new Object[]{table.getStorePath(), total, count, response.getInfo(), response.getReceiveTime()});
        }

        /**
         * 再次上传
         *
         * @param info 请求消息
         */
        private void uploadAgain(AbstractSdtsRequest info) {
            try {
                //若执行失败，则改操作类型为-变更，再次执行
                Method setMethod = info.getClass().getDeclaredMethod("setOperation", Integer.class);
                setMethod.invoke(info, UPDATE_OPERATION);
                uploadSdtsInfos(info);
            } catch (Exception e1) {
                LOGGER.error(e1.getMessage(), e1);
            }
        }

        /**
         * 标记上传错误的数据
         *
         * @param info 上传的数据
         */
        private void recordUploadFailture(AbstractSdtsRequest info) {
            try {
                Method getMethod = info.getClass().getDeclaredMethod("getId");
                String id = getMethod.invoke(info).toString();
                transferUserInfoDao.updateUploadFailture(id, table.getTableName());
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
        }

    }

    public static void main(String[] args) {
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        UploadRequestInfos service = context.getBean(UploadRequestInfos.class);
        if (args != null) {
            List<ExecuteTable> executeTables = Arrays.stream(args)
                    .filter(StringUtils::isNotBlank)
                    .map(storeName -> Arrays.stream(ExecuteTable.values())
                            .filter(table -> table.getStorePath().equalsIgnoreCase(storeName))
                            .findFirst()
                            .orElse(null))
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
            service.setExecuteTables(executeTables);
        }
        service.uploadReqInfos();
    }
}
